import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.lang.Console;
import cn.hutool.core.lang.Opt;
import cn.hutool.core.util.NumberUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;

/**
 * 计算员
 *
 * @author <achao1441470436@gmail.com>
 * @since 2021/11/4 0004 23:24
 */
public class Calculator extends KeyedProcessFunction<String, Order, HorseTicket> {
    public static final int NUMBER_LIMIT = 10;
    public static final int WEEK_DAY_LIMIT = 7;
    private static final long serialVersionUID = -6248614038880488977L;
    private transient ValueState<List<Score>> scoreBoard;

    @Override
    public void open(Configuration parameters) {
        // 限定类型，这里我是一个List<Score>
        ValueStateDescriptor<List<Score>> scoreDescriptor = new ValueStateDescriptor<>(
                "scoreboard",
                Types.LIST(TypeInformation.of(Score.class)));
        scoreBoard = getRuntimeContext().getState(scoreDescriptor);
    }

    @Override
    public void processElement(Order order, Context ctx, Collector<HorseTicket> out) throws Exception {
        List<Score> scores = Opt.ofNullable(this.scoreBoard.value()).orElseGet(CollUtil::newArrayList);
        Console.log(order);
        // 计分，先计算这次订单积分
        Double score = order.getGoods().parallelStream().map(Goods::getGoodsType).map(GoodsTypeEnum::getScore).reduce((double) 0, NumberUtil::add, NumberUtil::add);
        // 输出提示语
        Console.log("{},Your score is {} at {}.", order.getUsername(), score, DateUtil.formatDateTime(order.getCreateTime()));

        scores.add(Score.builder().score(score).time(order.getCreateTime()).build());
        // 如果一周内积够10分，送骑马劵，带顾客去马场体验骑马
        if (scores.parallelStream().filter(s -> DateUtil.between(s.getTime(), DateUtil.date(), DateUnit.DAY) <= WEEK_DAY_LIMIT).map(Score::getScore).reduce((double) 0, NumberUtil::add, NumberUtil::add) >= NUMBER_LIMIT) {
            driveHorse(out, order, scores.toArray(new Score[0]));
            return;
        }

        // 否则积分
        this.scoreBoard.update(scores);
    }

    private void driveHorse(Collector<HorseTicket> out, Order order, Score... scores) {
        out.collect(HorseTicket.builder().username(order.getUsername()).scores(Arrays.asList(scores)).build());
        this.scoreBoard.clear();
    }

}
